package com.saic.data.table;

import java.io.IOException;
import java.text.DecimalFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;

import com.saic.data.dotask.QueryByFromTo;
import com.saic.data.dotask.QueryByValue;
import com.saic.data.util.FileUtil;
import com.saic.data.util.HbaseUtil;

import it.sauronsoftware.ftp4j.FTPFile;

/**
 * @DESC:实用功能
 * @author Xinshiyou
 */
public class TableUtil {

	/** 配置表信息 */
	public static String configtab = "ftp_file";
	public static String streamchecktab = "month_streamno";
	public static String streamcheckfamily = "CF1";
	public static String configtabFamily = "FILE";
	public static String configCustomFamily = "CF1";
	public static long write2Hadfs_time = 60 * 1000;// write to HDFS
	
	/** 唯一性验证表格 */
	public static String tab_check = "unique_check";
	public static String flag_check = "isCheck";// custom10
	
	
	/**
	 * CSV_TMP/DONE table
	 */
	public static String getKey(TableFile tfile, int lineNumber) {
		return tfile.getRowkey()+ formatNum(lineNumber);
	}

	/**
	 * FTP table
	 */
	public static String getKey(Date createTime, String fileName) {
		return formatDate(createTime) + FileUtil.getMD5(fileName);
	}

	public static String formatDate(Date time) {
		return new SimpleDateFormat("yyyyMMddHHmmss").format(time);
	}

	public static String formatNum(int num) {
		return new DecimalFormat("00000000").format(num);
	}

	public static void pm(Class<?> cls, String msg) {
		pm(cls.getName(), msg);
	}

	public static void pm(String cls, String msg) {
		String data = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
		System.out.println("["+data+"] Print by " + cls + "-->" + msg);
	}
	
	public static Map<String, String> getFileMap(FTPFile f) {
		Map<String, String> map = new HashMap<String, String>();
		map.put("fileName", f.getName().split("\\.")[0]);
		map.put("fileType", FileUtil.getType(f.getName()));
		map.put("fileSize", f.getSize() + "");
		map.put("fileStats", "New");
		map.put("modifiedDate", formatDate(f.getModifiedDate()));
		map.put("checkTime", System.currentTimeMillis() + "");
		map.put("createTime", System.currentTimeMillis() + "");
		map.put("UUID", FileUtil.getUUID());
		return map;
	}
	public static Map<String,String> getCustomMap(TableFile tfile) {
		Map<String,String> map = new HashMap<String, String>();
		map.put("tableName", tfile.gettConf().getTablename());	
		map.put("streamNo", tfile.getStreamNo());
		map.put("financeYear",tfile.getFinanceYear());
		map.put("financeMonth",tfile.getFinanceMonth());
		map.put("financeDate", tfile.getFinanceDate());
		return map;
	}
	
	public static TableAction getAction(TableFile table,FTPFile f) throws ParseException, IOException, Exception{
		TableAction taction = null ;
		String fileDate = formatDate(f.getModifiedDate());
		pm("TableUitl-getAction", "scan configtable by fileName="+table.getFileName());
		ResultScanner rs = HbaseUtil.QueryByConditionValue(TableUtil.configtab, TableUtil.configtabFamily,
				"fileName", table.getFileName());
        
		for(Result r:rs){
			if (r.isEmpty()){
				taction = TableAction.New;
				rs.close();
				return taction;
			}
			//pm("TableUitl-getAction", "comparing key="+new String(r.getRow()));
			if(table.checkISExist(r)){
				String stats = HbaseUtil.getFromResult(r,configtabFamily,"fileStats");
				String tableDate = HbaseUtil.getFromResult(r,configtabFamily,"modifiedDate");
				long preFS = Long.parseLong(HbaseUtil.getFromResult(r,configtabFamily,"fileSize"));
				long preDT = Long.parseLong(HbaseUtil.getFromResult(r,configtabFamily,"checkTime"));
				if (stats.equals(TableAction.UpdateDB.toString())) {
					taction = TableAction.None;
					return taction;	
				}
				if (stats.equals(TableAction.UploadRecord.toString())){
					taction = TableAction.UploadRecord;
					return taction;	
				}
				if (stats.equals(TableAction.New.toString()) || stats.equals(TableAction.Wait.toString())) {
					if(f.getSize() != preFS ||System.currentTimeMillis() - preDT < write2Hadfs_time||f.getSize() ==0) 
						taction = TableAction.Wait;
					else {  
						if (table.checkStreamNo()) taction = TableAction.UploadRecord;
						else taction = TableAction.DeleteFile;
						}
				}else{
					if(tableDate.compareTo(fileDate) < 0 || f.getSize() != preFS) taction = TableAction.Wait;
					else  taction = TableAction.DeleteFile;	
				}
			}
		}
		rs.close();
		if (taction == null) taction=TableAction.New;
		return taction;	
		
	}
	
	public static void deleteByfileKey(String tableName,String fileKey) throws Exception {
		QueryByFromTo qb = new QueryByFromTo(tableName, fileKey + "00000000", fileKey + "99999999");
		qb.doTask();
		ResultScanner rs = qb.getResult();
		for(Result res :rs)
			HbaseUtil.deleteRow(tableName, res.getRow());
	}
}